package en.edu.zxj.mq.mqserver.datacenter;

import en.edu.zxj.mq.common.MqException;
import en.edu.zxj.mq.common.util.BinaryUtils;
import en.edu.zxj.mq.mqserver.core.MSGQueue;
import en.edu.zxj.mq.mqserver.core.Message;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.Contract;
import org.jetbrains.annotations.NotNull;

import java.io.*;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.LinkedList;
import java.util.List;
import java.util.Scanner;

/**
 * Created with IntelliJ IDEA.
 * Description：消息持久化存储
 * 存储单位是以队列名字为单位存储的
 *
 * @author: zxj
 * @date: 2024-02-28
 * @time: 13:43:23
 */
@Slf4j
public class MessageFileManger {

    private final static String BASIC_DIR = "./data/";

    /**
     * @description: 内部类, 用于管理 queue_stat.txt 中的数据
     * 存储格式: totalCount \t validCount
     * 作用: 为了后面垃圾回收功能做准备的
     * 约定: 当 有效信息的占比低于 50% 时, 并且 总的消息数目大于 2000 时, 触发 GC 功能
     **/
    static public class Stat {
        // 总信息的存储数目
        public Integer totalCount;
        // 有效的信息数目
        public Integer validCount;


        // 最少消息数目
        private static final Integer atLeastCount = 2000;
        // 最低有效信息占比
        private static final Double minProportion = 0.5;

    }

    public void init() {
        // 暂时不需要任何初始化操作, 方便后续扩展
    }

    // 设定信息存储的目录和文件

    /**
     * @description: 用来获取指定队列信息存储的目录
     **/
    @Contract(pure = true)
    private @NotNull String getQueueDir(String queueName) {
        return BASIC_DIR + queueName;
    }

    /**
     * @description: 用来获取指定队列信息数据存储路径
     **/
    @Contract(pure = true)
    private @NotNull String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.txt";
    }

    /**
     * @description: 用来获取指定队列信息记录存储路径
     **/
    @Contract(pure = true)
    private @NotNull String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    /**
     * @description: 用来获取指定队列新数据存储路径
     **/
    @Contract(pure = true)
    private @NotNull String getQueueDataNewPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }


    /**
     * @description: 读取 queue_stat.txt 文件里面的内容
     **/
    private @NotNull Stat readStat(String queueName) throws IOException {
        Stat stat = new Stat();
        try (InputStream inputStream = Files.newInputStream(Paths.get(getQueueStatPath(queueName)))) {
            // 因为 queue_stat.txt 里面存储的内容是文本的, 所以可以使用 Scanner 来读取
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();
            stat.validCount = scanner.nextInt();
        }
        return stat;
    }

    /**
     * @description: 向 queue_stat.txt 文件里面写入内容
     **/
    private void writeStat(String queueName, @NotNull Stat stat) throws IOException {
        // 使用 PrintWrite 来写文件.
        // OutputStream 打开文件, 默认情况下, 会直接把原文件清空. 此时相当于新的数据覆盖了旧的.
        try (OutputStream outputStream = Files.newOutputStream(Paths.get(getQueueStatPath(queueName)))) {
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();
        }
    }


    /**
     * @description: 创建相关目录信息, 并进行相关的初始化操作
     **/
    public void createQueueFiles(String queueName) throws IOException {
        // 1. 创建对应的目录
        File fileDir = new File(getQueueDir(queueName));
        if (!fileDir.exists()) {
            // 不存在对应目录
            if (!fileDir.mkdirs()) {
                throw new IOException("创建目录失败, fileDir: " + fileDir.getAbsoluteFile());
            }
        }
        // 2. 创建 queue_data.txt 文件
        File fileData = new File(getQueueDataPath(queueName));
        if (!fileData.exists()) {
            // 不存在对应文件
            if (!fileData.createNewFile()) {
                throw new IOException("创建目录失败, fileData: " + fileData.getAbsoluteFile());
            }
        }
        // 3. 创建 queue_stat.txt 文件
        File fileStat = new File(getQueueDataPath(queueName));
        if (!fileStat.exists()) {
            // 不存在对应文件
            if (!fileStat.createNewFile()) {
                throw new IOException("创建目录失败, fileStat: " + fileStat.getAbsoluteFile());
            }
        }
        // 4. 初始化 queue_stat.txt 文件
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName, stat);
    }


    /**
     * @description: 销毁消息的目录文件
     **/
    public void destroyQueueFiles(String queueName) throws IOException {
        // 先删除文件, 在删除目录
        File fileData = new File(getQueueDataPath(queueName));
        boolean ok1 = fileData.delete();
        File fileStat = new File(getQueueStatPath(queueName));
        boolean ok2 = fileStat.delete();
        File fileDir = new File(getQueueDir(queueName));
        boolean ok3 = fileDir.delete();
        if (!ok1 || !ok2 || !ok3) {
            // 但凡有一个失败, 就算整体是失败的
            throw new IOException("删除指定文件和目录失败, dir: " + fileDir.getAbsoluteFile());
        }
    }


    /**
     * @description: 判断 queueName 对应的文件是否存在
     * 比如后续生产者给 broker server 生产消息了, 这消息可能需要被记录到文件上(取决于该信息是否需要持久化)
     * @return:
     **/
    public boolean checkFilesExists(String queueName) {
        // 判断队里的数据文件和状态文件是否存在即可
        File fileData = new File(getQueueDataPath(queueName));
        File fileStat = new File(getQueueStatPath(queueName));
        return fileStat.exists() && fileData.exists();
    }


    /**
     * @description: 增
     * 将 message 放到 msgQueue 对应的队列文件中
     * @param: [msgQueue 消息队列, message 需要存储的信息 - 内存中也会管理该对象]
     **/
    public void sendMessage(@NotNull MSGQueue msgQueue, Message message) throws MqException, IOException {
        // 1. 检查当前队列要写入的文件是否存在
        if (!checkFilesExists(msgQueue.getName())) {
            throw new MqException("[MessageFileManager] 队列所对应的文件不存在! queueName=" + msgQueue.getName());
        }

        // 2. 把 message 转化为 字节数组
        byte[] messageBinary = BinaryUtils.toBytes(message);

        // 将 messageBinary 写入到 msgQueue 所对应的队列文件中
        // 文件属于一个公共资源, 此时进行写操作, 存在线程安全的问题
        // 需要对对应的队列进行加锁, 确保同时向同一个队列中写入信息是线程安全的
        synchronized (msgQueue) {
            // 3. 设置 Message 对象中 offsetBeg 和 offsetEnd 字段
            // 3.1 获取此时对应文件的总长度, fileQueueData.length() 就可以获取
            File fileQueueData = new File(getQueueDataPath(msgQueue.getName()));
            // 3.2 计算
            // 把新的 message 写入到文件中, offsetBeg = 旧的总长度 + 4, offsetEnd = 旧的总长度 + messageBinary.length + 4
            message.setOffsetBeg(fileQueueData.length() + 4);
            message.setOffsetEnd(fileQueueData.length() + messageBinary.length + 4);

            // 4. 将 messageBinary 写入到文件的默认, 注意: 这里是追加写
            try (OutputStream outputStream = new FileOutputStream(getQueueDataPath(msgQueue.getName()), true)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    // 4.1 先写入新 messageBinary 的长度 -- 固定占四个字节
                    // 知识点: outputStream.write() 参数看似是 int 类型, 但是实际上只是写入一个字节的数据, dataOutputStream.writeInt() 就是写四个字节的数据
                    dataOutputStream.writeInt(messageBinary.length);
                    // 4.2 写入主体信息
                    dataOutputStream.write(messageBinary);
                }
            }

            // 5. 更新信息统计文件的信息
            Stat stat = readStat(msgQueue.getName());
            stat.validCount += 1;
            stat.totalCount += 1;
            writeStat(msgQueue.getName(), stat);
        }
    }

    /**
     * @description: 删除 message
     * 这里的删除是逻辑删除, 也就是把硬盘上存储的这个数据里面的那个 isValid 属性, 设置为 0x0
     * 1. 先把这个文件中的这一段数据, 读出现来, 还原回 Message 对象
     * 2. 把 isValid 该成 0;
     * 3. 把上述数据重新写回文件
     * 此处这个参数中的 message 对象, 必须要包含有效的 offsetBeg 和 offsetEnd
     **/
    public void deleteMessage(@NotNull MSGQueue msgQueue, @NotNull Message message) throws IOException, ClassNotFoundException {
        // 修改文件, 存在线程安全问题
        synchronized (msgQueue) {
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(msgQueue.getName()), "rw")) {
                // 1. 先从文件中读取对应的 message 数据
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.read(bufferSrc);
                // 2. 将当前读出来的二进制数据, 转换成 Message 对象
                Message diskMessage = (Message) BinaryUtils.fromBytes(bufferSrc);
                // 3. 把 isValid 设置为无效
                diskMessage.setIsValid((byte) 0x0);
                // 此处不需要宰割参数这个 Message 的 isValid 设为 0, 因为这个参数代表的内容中管理的 Message 对象, 而这个对象也马上要被从内存中销毁了
                // 4. 重新写入文件
                byte[] buffDest = BinaryUtils.toBytes(diskMessage);
                // 虽然上面已经 seek 过了, 但是上面 seek 完了之后, 进行了读操作, 这一读, 就导致, 文件光标往后移动, 移动到下一个信息的位置了,
                // 因此想要接下来的写入, 能能够刚好写回到之前的位置, 就需要重新调整文件光标
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.write(buffDest);
                // 通过上述这些操作, 对于文件来说, 只是有一个字节发生了改变了而已~
            }
            // 更新统计文件, 把一个消息设置为无效了, 此时有效信息个数就需要 -1
            Stat stat = readStat(msgQueue.getName());
            if (stat.validCount > 0) {
                stat.validCount -= 1;
            }
            writeStat(msgQueue.getName(), stat);
        }
    }


    /**
     * @description: 查
     * 使用这个方法, 从文件中, 读取所有的消息内容, 加载到内存中 (具体来说是放到一个链表里面)
     * 这个方法,使用一个 LinkedList, 主要目的是为了后续进行头删操作
     * 这个方法的参数, 只是一个 queueName 而不是 MSGQueue 对象, 因为这个方法不需要加锁,只使用 queueName 就够了
     * 由于该方法是在程序启动调用, 此时服务还不能处理请求, 不涉及多线程操作文件
     **/
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try (InputStream inputStream = Files.newInputStream(Paths.get(getQueueDataPath(queueName)))) {
            try (DataInputStream dataInputStream = new DataInputStream(inputStream)) {
                // 这个变量记录当前文件的光标
                long currentOffset = 0;
                while (true) {
                    // 1. 先读取当前消息长度字段
                    int messageSize = dataInputStream.readInt();
                    // 2. 按照这个长度, 读取后续的数据长度
                    byte[] buffSrc = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffSrc);
                    if (messageSize != actualSize) {
                        // 如果不匹配, 说明文件有问题, 格式错了
                        throw new MqException("[MessageFileManager] 文件格式错误, queueName=" + queueName);
                    }
                    // 3. 把读到的二进制数据, 反序列化为 Message 对象
                    Message message = (Message) BinaryUtils.fromBytes(buffSrc);
                    // 4. 判定一下, 看看这个消息对象, 是不是无效对象
                    if (message.getIsValid() != 0x1) {
                        // 无效数据, 直接跳过
                        // 虽然消息是无效数据, 但是 offset 仍要更新
                        currentOffset += (4 + messageSize);
                        continue;
                    }
                    // 5. 有效数据, 则需要将这个 Message 对象加入到链表中, 加入之前还需要添加 OffsetBeg 和 OffsetEnd
                    //    进行计算 Offset 的时候, 需要当前文件光标的位置, 由于当下使用的 DataInputStream 并不方便直接获取文件光标位置
                    //    因此需要手动计算下文件光标
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageSize);
                    currentOffset += (4 + messageSize);
                    messages.add(message);
                }
            } catch (EOFException e) {
                // 这个异常是表示读取到文件的末尾了, 这是 DataInputStream 中规定的
                // 不需要做什么特殊处理
                log.info("恢复磁盘文件数据完成");
            }
        }
        return messages;
    }


    /**
     * @description: 检查当前是否需要针对队列的消息数据文件进行 GC
     **/
    public boolean checkGC(String queueName) throws IOException {
        // 判断是否要 GC, 是根据总消息数和有效消息数, 这两个值都是在 消息统计文件中的
        Stat stat = readStat(queueName);
        return stat.totalCount > Stat.atLeastCount && (double) stat.validCount / (double) stat.totalCount < Stat.minProportion;
    }


    /**
     * @description: 垃圾回收, 防止存储过多垃圾信息
     * 通过这个方法, 真正执行消息数据文件的垃圾回收操作
     * 使用复制算法来完成
     * 创建一个新的文件, 名字就是 queue_data_new.txt
     * 把之前消息数据文件中的有效消息都读出来, 写到新的文件中
     * 删除旧的文件,在把新的文件改名回 queue_data.txt
     * 同时要更新消息统计文件
     **/
    public void gc(MSGQueue msgQueue) throws MqException, IOException, ClassNotFoundException {
        // 进行 gc 的过程, 是针对消息数据文件进行大洗牌, 这个过程中, 其他线程不能针对该队列的消息文件做任何修改
        synchronized (msgQueue) {
            // 由于 gc 操作可能比较耗时, 此处统计一下消耗时间
            long gcBeg = System.currentTimeMillis();

            // 1. 创建一个新的文件
            File queueDataNewFile = new File(getQueueDataNewPath(msgQueue.getName()));
            if (queueDataNewFile.exists()) {
                // 正常情况下, 这个文件不应该存在, 如果存在, 就是意外, 说明上次 gc 了一半, 程序意味崩溃了
                throw new MqException("[MessageFileManager] gc 时发现该队列的 queue_data_new 已经存在, queueName=" + msgQueue.getName());
            }
            boolean ok = queueDataNewFile.createNewFile();
            if (!ok) {
                throw new MqException("[MessageFileManager] 创建文件失败 ,queueDataNewFile=" + queueDataNewFile.getAbsoluteFile());
            }

            // 2. 从旧文件中, 读取出所有的有效消息对象
            LinkedList<Message> messages = loadAllMessageFromQueue(msgQueue.getName());

            // 3. 把有效信息写入到新的文件中
            try (OutputStream outputStream = Files.newOutputStream(Paths.get(getQueueDataNewPath(msgQueue.getName())))) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    for (Message message : messages) {
                        byte[] buffer = BinaryUtils.toBytes(message);
                        // 先写长度
                        dataOutputStream.writeInt(buffer.length);
                        // 在写整体数据
                        dataOutputStream.write(buffer);
                    }
                }
            }


            // 4. 删除旧的数据文件, 并把新的文件进行重命名
            File queueDataOldFile = new File(getQueueDataPath(msgQueue.getName()));
            ok = queueDataOldFile.delete();
            if (!ok) {
                throw new MqException("[MessageFileManager] 删除旧的数据文件失败, queueDataOldFile: " + queueDataOldFile.getAbsoluteFile());
            }
            // 把 queue_data_new.txt => queue_data.txt
            ok = queueDataNewFile.renameTo(queueDataOldFile);
            if (!ok) {
                throw new MqException("[MessageFileManager] 文件重命名失败, queueDataOldFile: " + queueDataOldFile.getAbsoluteFile() + ", queueDataNewFile: " + queueDataNewFile.getAbsoluteFile());

            }

            // 5. 更新统计文件
            Stat stat = readStat(msgQueue.getName());
            stat.totalCount = messages.size();
            stat.validCount = messages.size();

            long gcEnd = System.currentTimeMillis();
            log.info("gc 执行消耗时间: {} ms", (gcEnd - gcBeg));
        }
    }


}













